package com.yootk.rockemq.orders;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.List;

public class OrdersMessageOneWayProducer { // 消息生产者
    // 如果此时你使用的是集群服务，则每个主机之间使用“,”分割
    public static final String NAME_SERVER_LIST = "rocketmq-server:9876"; // NameServer
    public static final String PRODUCER_GROUP = "muyan-group-orders"; // 定义消费组
    public static final String TOPIC = "TopicYootk"; // 定义主题名称
    public static final String ACCESS_KEY = "RocketMQMuyan"; // 定义用户名
    public static final String SECRET_KEY = "helloyootk"; // 定义密码
    public static final String TAGS[] = new String[] {"tagA", "tagB", "tagC", "tagD", "tagE"}; // 业务标签
    public static void main(String[] args) throws Exception { // 懒人必备的处理形式
        // 1、由于此时的RocketMQ启动了ACL安全认证的保护机制，所以需要配置相应的回调
        RPCHook clientHook = new AclClientRPCHook(new SessionCredentials(ACCESS_KEY, SECRET_KEY));
        // 2、创建一个消息的生产者
        DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP, clientHook);
        producer.setNamesrvAddr(NAME_SERVER_LIST); // NameServer地址
        producer.start(); // 启动生产者
        for (int x = 0; x < 10; x++) {// 3、采用循环的方式进行消息的生产
            Message msg = new Message(TOPIC, TAGS[x % TAGS.length], ("【" + String.format("%03d", x) + "】" + "沐言科技：www.yootk.com")
                    .getBytes(RemotingHelper.DEFAULT_CHARSET)); // 消息需要二进制
            producer.sendOneway(msg, new MessageQueueSelector() {// 单向消息
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    int index = ((Integer) arg) % mqs.size(); // 计算队列
                    return mqs.get(index);
                }
            }, x % TAGS.length); // 设置一个处理的参数
        }
        producer.shutdown(); // 关闭生产者
    }
}